package org.rency.mq.activemq.listener.failover.impl;

import com.alibaba.fastjson.JSON;
import org.rency.mq.activemq.constant.DeliveryMode;
import org.rency.mq.activemq.constant.DestinationType;
import org.rency.mq.activemq.constant.MessageFormat;
import org.rency.mq.activemq.jms.service.JmsAccessor;
import org.rency.mq.activemq.listener.failover.FailoverService;
import org.rency.mq.activemq.request.MQRequest;
import org.rency.mq.activemq.utils.JsonMessageUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.JMSException;
import javax.jms.Session;
import java.util.LinkedHashMap;
import java.util.Map;

/**
 * ${todo}
 *
 * @author: Created by user_rcy@163.com on 2016/10/20.
 * @version: V1.0.0
 * @since JDK 1.8
 */
public class FailoverServiceImpl implements FailoverService {

    private boolean isSendFail = true;
    private boolean isSendRetry = true;
    private static Logger logger = LoggerFactory.getLogger(FailoverServiceImpl.class);
    private JmsAccessor mqAccessor;

    public void setMqAccessor(JmsAccessor mqAccessor) {
        this.mqAccessor = mqAccessor;
    }

    public void retry(MQRequest request) {
        Map<String, Object> props = request.getProperties();
        int failedTimes = (Integer) props.get("failedTimes");
        // 为-1时直接失败.
        if (-1 != failedTimes) {
            props.put("failedTimes", ++failedTimes);
        }

        int retryTimes = (Integer) props.get("retryTimes");

        // 如果失败次数未达到重试次数，则将请求重新放到MQ的对应队列中.
        if (-1 != failedTimes && failedTimes <= retryTimes) {
            sendToRetryQueue((String) props.get("retryQueue"), request);
        } else {        //超过最大失败次数后将消息序列化为JSON后发送到失败队列
            sendToFailedQueue((String) props.get("failedQueue"), request);
        }
    }

    private void sendToRetryQueue(final String queue,final MQRequest request) {
        if (!isSendRetry) {
            logger.warn("没有配置发送到重试队列.消息将被丢弃.message:" + request.toString());
            return;
        }
        logger.warn("Retry to send MQRequest: " + request.toString());

        try {
            mqAccessor.sendMessage(request.getDestination(), request.getDestinationType(), request);
        } catch (Exception e) {
            logger.error("Send message to original queue fail.", e);
            sendToFailedQueue(null, request);
        }
    }

    private void sendToFailedQueue(String destination,MQRequest request) {
        if (!isSendFail) {
            logger.warn("没有配置发送到失败队列.消息将被丢弃.message:" + request.toString());
            return;
        }
        logger.warn("Send to failed queue: " + request.toString());
        if(destination == null){
            Map<String, Object> props = request.getProperties();
            destination = (String) props.get("failedQueue");
        }

        try {
            mqAccessor.sendMessage(destination, DestinationType.QUEUE, MessageFormat.OBJECT, false,
                    Session.AUTO_ACKNOWLEDGE, DeliveryMode.PERSISTENT,JSON.toJSONString(request));
        } catch (JMSException ex) {
            logger.error("Send message to failed queue failed.", ex);
        }
    }

    public void retry(Object request, int retryTimes,String retryQueue, String failedQueue) {
        logger.info("Retry messge: " + request);

        if (request instanceof MQRequest) {
            MQRequest mqRequest = (MQRequest)request;

            if (!mqRequest.containsProperty("retryQueue")) {
                mqRequest.addProperty("retryTimes", retryTimes);
                mqRequest.addProperty("retryQueue", retryQueue);
                mqRequest.addProperty("failedQueue", failedQueue);
                mqRequest.addProperty("failedTimes", 0);
            }

            retry(mqRequest);
        } else if (request instanceof String) {
            retryString((String) request, retryTimes, retryQueue, failedQueue);
        }else{
            logger.error("Unsupported request type. Request: " + request);
        }
    }

    @SuppressWarnings("unchecked")
    private void retryString(String request, int retryTimes,String retryQueue,String failedQueue) {
        LinkedHashMap<String, Object> map = null;
        try {
            map = (LinkedHashMap<String, Object>) JSON.parseObject((String) request, Object.class);
        } catch (Exception e) {
            logger.error("Unmarshall message failed. Message may not json type. Request: " + request);

            throw new UnsupportedOperationException(e);
        }

        int deliveryMode = JsonMessageUtils.getDeliveryMode(map);
        int acknowledgeMode = JsonMessageUtils.getAcknowledgeMode(map);

        boolean transcated = false;

        if(map.containsKey("transacted")){
            transcated = (Boolean) map.get("transacted");
        }else{
            if(acknowledgeMode == Session.SESSION_TRANSACTED){
                transcated = true;
            }
        }

        String destination = getDestination(map, retryTimes, retryQueue,
                failedQueue);

        try {
            request = JSON.toJSONString(map);
        } catch (Exception e) {
            e.printStackTrace();
        }

        try {
            mqAccessor.sendMessage(destination, DestinationType.QUEUE, MessageFormat.JSON, transcated,acknowledgeMode, deliveryMode, request);
        } catch (JMSException e) {
            logger.error("Retry failed. Request: " + request);
        }
    }

    @SuppressWarnings("unchecked")
    private String getDestination(Map<String, Object> map, int retryTimes,String retryQueue, String failedQueue) {
        Map<String, Object> props = (Map<String, Object>)map.get("properties");

        if(props == null){
            props = new LinkedHashMap<String, Object>();
            map.put("properties", props);
        }

        if(!props.containsKey("retryTimes")){
            props.put("retryTimes", retryTimes);
            props.put("retryQueue", retryQueue);
            props.put("failedTimes", 0);
            props.put("failedQueue", failedQueue);
        }

        return getDestination(map);
    }

    private String getDestination(Map<String, Object> map){
        Map<String, Object> props = (Map<String, Object>)map.get("properties");

        int failedTimes = (Integer) props.get("failedTimes");
        // 为-1时直接失败.
        if (-1 != failedTimes) {
            props.put("failedTimes", ++failedTimes);
        }

        int retryTimes = (Integer) props.get("retryTimes");

        // 如果失败次数未达到重试次数，则将请求重新放到MQ的对应队列中.
        if (-1 != failedTimes && failedTimes <= retryTimes) {
            return (String)props.get("retryQueue");
//			return (String)map.get("destination");
        } else {
            return (String)props.get("failedQueue");
        }
    }

    public boolean isSendFail() {
        return isSendFail;
    }

    public void setSendFail(boolean isSendFail) {
        this.isSendFail = isSendFail;
    }

    public boolean isSendRetry() {
        return isSendRetry;
    }

    public void setSendRetry(boolean isSendRetry) {
        this.isSendRetry = isSendRetry;
    }

}